package com.bclz.nio.simpleReactor;


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

/**
 * Reactor Acceptor Handle设计模式实现
 */
public class Reactor2 implements Runnable {

    private ServerSocketChannel serverSocketChannel;
    private Selector selector;

    private Reactor2() {}

    public static Reactor2 build(int port) throws IOException {
        Reactor2 reactor = new Reactor2();
        reactor.selector = Selector.open();
        reactor.serverSocketChannel = ServerSocketChannel.open();
        reactor.serverSocketChannel.bind(new InetSocketAddress("localhost", port));
        reactor.serverSocketChannel.configureBlocking(false);
        new Acceptor(reactor.selector,reactor.serverSocketChannel);
        return reactor;
    }

    @Override
    public void run() {
        try {
            while(!Thread.interrupted()){
                int select = selector.select(100);
                if(select<1){
                    continue;
                }
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    dispatch(selectionKey);
                    iterator.remove();
                }
            }
        } catch (IOException e) {
            System.out.println("Dispatch error...");
        }
    }

    private void dispatch(SelectionKey selectionKey){

        Runnable runnable = (Runnable) selectionKey.attachment();
        if(runnable!=null){
            runnable.run();
        }

    }
}


class Acceptor implements Runnable{

    private ServerSocketChannel serverSocketChannel;
    private Selector selector;

    public Acceptor(Selector selector,ServerSocketChannel serverSocketChannel) throws ClosedChannelException {
        SelectionKey selectionKey = serverSocketChannel.register(selector,
                SelectionKey.OP_ACCEPT);
        selectionKey.attach(this);
        this.serverSocketChannel=serverSocketChannel;
        this.selector=selector;
    }

    @Override
    public void run() {
        try {
            System.out.println("new connect accept...");
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            new Handle(selector,socketChannel);
        } catch (IOException e) {
            System.out.println("socket accept error...");
        }
    }
}



class Handle implements Runnable{

    private static int READ=0;
    private static int WRITE=1;

    private SocketChannel socketChannel;
    private SelectionKey selectionKey;
    private int state;
    private ByteBuffer readBuffer=ByteBuffer.allocate(1024);
    private ByteBuffer writeBuffer=ByteBuffer.allocate(1024);




    public Handle(Selector selector,SocketChannel socketChannel) throws ClosedChannelException {
        SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_READ);
        selectionKey.attach(this);
        this.socketChannel=socketChannel;
        this.selectionKey= selectionKey;
        this.state=READ;
    }

    @Override
    public void run() {
        try{
            if(state==READ){
                read();
            }else{
                write();
            }
        }catch (Exception e){
            System.out.println();

        }

    }

    private void write() throws IOException {
        System.out.println("WRITE...");
        int rows=0;
        while(writeBuffer.hasRemaining()){
            rows = socketChannel.write(writeBuffer);
            if(rows==0){ //系统socket缓冲区已满
                break;
            }
        }
        writeBuffer.compact();
        if(rows>0){
            writeBuffer.clear();
            //取消写事件，防止空转
            selectionKey.interestOps(selectionKey.interestOps()&~SelectionKey.OP_WRITE);
            state=READ;
            System.out.println("write end...");
        }

    }


    private void read() throws IOException {
        System.out.println("read...");
        StringBuilder sb=new StringBuilder();
        int count;
        while((count=socketChannel.read(readBuffer))>0){
            sb.append(new String(readBuffer.array(),"utf-8"));
            readBuffer.clear();
        }
        if(count==-1){
            closeSocketChannel(selectionKey);
            return;
        }
        System.out.println("already review data :"+sb.toString());
        writeBuffer= ByteBuffer.wrap("already review data !!!!".getBytes());
        selectionKey.interestOps(selectionKey.interestOps()|SelectionKey.OP_WRITE);
        state=WRITE;
    }

    private void closeSocketChannel(SelectionKey selectionKey){
        try {

            selectionKey.cancel();
            selectionKey.channel().close();
            System.out.println(selectionKey.hashCode()+" already close...");
        } catch (IOException ex) {
            System.out.println("close channel error...");
        }


    }
}